package xfld;

import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2022/3/23
 */
public class kafkaJoinMysql {
    public static void main(String[] args) {
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        streamExecutionEnvironment.setStateBackend(new HashMapStateBackend());
        streamExecutionEnvironment.enableCheckpointing(10000L);
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage("hdfs://172.17.1.102:8020/checkpoint/");
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(streamExecutionEnvironment);

        streamTableEnvironment.executeSql("create table tmp_stg_xfld_toilet_data (`id` string, `version` INTEGER, `timestamp` BIGINT, " +
                "`state` MAP<String,String>, `quality` MAP<String,String>, `upTIMESTAMP` MAP<String,BIGINT>)" +
                "with ('connector' = 'kafka','topic' = 'ecityos-iot-msg-data','properties.bootstrap.servers' = " +
                "'172.17.222.120:9092','properties.group.id' = 'test7Group','scan.startup.mode' = 'earliest-offset','format' = 'json')");

        streamTableEnvironment.executeSql("create table ywgl_device (" +
                " `id` int ," +
                "  `device_code` varchar(50)," +
                "  `iot_code` varchar(255) ," +
                "  `device_name` varchar(255)," +
                "  `device_spec` varchar(255)," +
                "  `category_code` varchar(50)," +
                "  `location_code` varchar(50) ," +
                "  `location_name` varchar(255) ," +
                "  `use_department_code` varchar(50) ," +
                "  `manage_department_code` varchar(50) ," +
                "  `manage_person_name` varchar(255) ," +
                "  `supplier_code` varchar(50) ," +
                "  `special_status` int," +
                "  `barcode` varchar(50) ," +
                "  `prduce_date` TIMESTAMP," +
                "  `contract_no` varchar(50) ," +
                "  `assets_code` varchar(50) ," +
                "  `assets_cost` decimal(20,2)," +
                "  `enable_date` TIMESTAMP," +
                "  `status` int ," +
                "  `running_status` varchar(2) ," +
                "  `start_status` varchar(2) ," +
                "  `parent_code` varchar(50) ," +
                "  `creator` varchar(64) ," +
                "  `create_date` TIMESTAMP," +
                "  `updator` varchar(64) ," +
                "  `update_date` TIMESTAMP," +
                "  `system_rule_code` varchar(50) ," +
                "  `hpid` varchar(50) ," +
                "  `map_display` int," +
                "  `three_dim_display` int ," +
                "  `map_display_level` int ," +
                "  `is_maintenance` int ," +
                "  `extended` varchar(50) ," +
                "  `virtual` int ," +
                "  `service_area_code` varchar(50) ," +
                "  `discipines_code` varchar(50) ," +
                "  `energy_code` varchar(50) ," +
                "  `spring_el` varchar(50) ," +
                "  `qr_code` varchar(50) ," +
                "  `sort_index` int," +
                "  `bim_custom_angle` varchar(1000) ," +
                "  `bim_file_name` varchar(100) ," +
                "  `model_id` varchar(50)," +
                "  `bim_element_id` varchar(100)," +
                "  `bim_position_low` varchar(255) ," +
                "  `bim_position_high` varchar(255) ," +
                "  `position_2d` varchar(255) ," +
                "  `position_3d` varchar(2000) ," +
                "  `object_id` varchar(255) ," +
                "  `real_data` varchar(255) ," +
                "  `arrt_value` varchar(255) ," +
                "  `supplier` varchar(255) ," +
                "  `recent_alarm_id` bigint," +
                "  `recent_alarm_desc` varchar(255) ," +
                "  `recent_alarm_time` TIMESTAMP," +
                "  `offset` varchar(1000) ," +
                "  `config` varchar(2000) ," +
                "  `remark` varchar(1000) ," +
                "  PRIMARY KEY (`id`) NOT ENFORCED) with (" +
                "'connector' = 'jdbc', 'url' = 'jdbc:mysql://172.17.222.168:3306/ywgl-xfld-dev','table-name' = 'ywgl_device'," +
                "'username' = 'root', 'password' = 'supconit', 'lookup.cache.max-rows' = '5000', 'lookup.cache.ttl' = '60000')");
        Table table = streamTableEnvironment.sqlQuery("select a.* from tmp_stg_xfld_toilet_data as a LEFT JOIN ywgl_device as b on a.`id`= b.`device_code` " +
                "where b.`category_code` = 'TOILET_TOILET'");
        table.execute().print();
    }
}
